/**
 * Software License, Version 1.0
 * 
 * Copyright 2003 The Trustees of Indiana University. All rights reserved.
 * 
 * 
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 * 
 * 1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license; 2) All redistributions in binary form
 * must reproduce the above copyright notice, this list of conditions and the
 * disclaimer listed in this license in the documentation and/or other materials
 * provided with the distribution; 3) Any documentation included with all
 * redistributions must include the following acknowledgement:
 * 
 * "This product includes software developed by the Community Grids Lab. For
 * further information contact the Community Grids Lab at
 * http://communitygrids.iu.edu/."
 * 
 * Alternatively, this acknowledgement may appear in the software itself, and
 * wherever such third-party acknowledgments normally appear.
 * 
 * 4) The name Indiana University or Community Grids Lab or NaradaBrokering,
 * shall not be used to endorse or promote products derived from this software
 * without prior written permission from Indiana University. For written
 * permission, please contact the Advanced Research and Technology Institute
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202. 5) Products
 * derived from this software may not be called NaradaBrokering, nor may Indiana
 * University or Community Grids Lab or NaradaBrokering appear in their name,
 * without prior written permission of ARTI.
 * 
 * 
 * Indiana University provides no reassurances that the source code provided
 * does not infringe the patent or any other intellectual property rights of any
 * other entity. Indiana University disclaims any liability to any recipient for
 * claims brought by any other entity based on infringement of intellectual
 * property rights or otherwise.
 * 
 * LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO
 * WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 * NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF
 * INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS.
 * INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS",
 * "VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.
 * LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR
 * ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION
 * GENERATED USING SOFTWARE.
 */
package cgl.narada.transport;

import java.util.Enumeration;
import java.util.Hashtable;
import java.util.Properties;
import java.util.StringTokenizer;

import cgl.narada.performance.MonitoringDataEvaluator;
import cgl.narada.performance.MonitoringService;
import cgl.narada.performance.MonitoringServiceException;
import cgl.narada.performance.MonitoringServiceImpl;
import cgl.narada.performance.PerformanceMeasurement;
import cgl.narada.protocol.Handler;
import cgl.narada.protocol.ProtocolHandler;
import cgl.narada.service.qos.impl.CommunicationsService;
import cgl.narada.transport.http.HTTPLinkFactory;
import cgl.narada.transport.https.HTTPSLinkFactory;
import cgl.narada.transport.ipsec.IPSecLinkFactory;
import cgl.narada.transport.multicast.MulticastLinkFactory;
import cgl.narada.transport.niotcp.NIOTCPLinkFactory;
import cgl.narada.transport.ptcp.PTCPLinkFactory;
import cgl.narada.transport.rtp.RTPLinkFactory;
import cgl.narada.transport.ssl.SSLLinkFactory;
import cgl.narada.transport.tcp.TCPLinkFactory;
import cgl.narada.transport.threadpool.PoolTCPLinkFactory;
import cgl.narada.transport.udp.UDPLinkFactory;
import cgl.narada.transport.up2p.UP2PLinkFactory;

/**
 * This class provides an implementation of the transport handler interface. The
 * transport handler interface serves as an intermediary between the protocol
 * layers & the underlying transport implementations. The transport handler
 * implementation pre-loads several link factory implementations. The
 * implementation provides all the functionalities that needs to be addressed
 * while implementing the transport handler interface. Among the operations
 * supported by the transport handler interface are the ability to
 * <ul>
 * <li>Load transport protocols for a certain type.
 * <li>Return information pertaining to the communication protocols that are
 * supported.
 * <li>Set up a communication <i>link</i> with the specified underlying
 * transport protocol, hostname, port num and other transport related factors.
 * <li>Setting up alternate links to a given node.
 * <li>Send data over a given link.
 * <li>Send data of a certain <i>type</i> and deploying the appropriate link
 * to facilitate data interchange.
 * <li>Set the link performance monitoring intervals.
 * <li>Shutdown communication services over a certain link.
 * <li>Shutdown communication services over all links to a giev node.
 * <li>Shutting down communication services over all links hosted by a given
 * node and releasing communication resources associated with the hosting node
 * </ul>
 * 
 * <p>
 * For the purposes of clarity in our dicussions let us assume that the
 * connection is being initiated by a node A to another node B. There are two
 * independent factors that need to borne in mind while using LinkFactory. The
 * first is of course setting the link factory properties. The second is
 * specifying the properties to create a link to some other node B.
 * 
 * <p>
 * To initialize the TCPLinkFactory one needs to specify the <b>TCPServerPort
 * </b>. When the value of this variable is set to <b>0</b> it implies that the
 * node A initiating a connection to node B will not accept link creation
 * requests from any other node. <br>
 * To create a TCP link to node B. The following properties needs to be
 * specified.
 * <li><b>hostname</b>: This is the hostname on which the node B's process is
 * running.
 * <li><b>portnum</b>: This is the port number on which node B accepts link
 * creation requests from other nodes. In other words, the <i>TCPServerPort </i>
 * specified to the TCPChannelFactory at node B is equal to portnum.
 * 
 * <p>
 * To initialize the UDPLinkFactory one needs to specify <b>UDPListenerPort
 * </b>, this is the port on which the process running this factory would listen
 * to for datagram packet communications. <br>
 * To create a UDP link to node B. The following properties needs to be
 * specified.
 * <li><b>hostname</b> - The host on which node B's process is running
 * <li><b>portnum</b> - The portnum on which node B is listening to, for
 * datagram packets.
 * 
 * <p>
 * No properties need to be setup during the initialization of the
 * MulticastLinkFactory. To enable receipt and sending of data to a given
 * multicast group the user needs to specify the <b>MulticastHost</b> (which
 * must be a valid multicast address) and <b>MulticastPort</b> information.
 * 
 * @author Shrideep Pallickara $Date$ $Revision$
 */

public class TransportHandlerImpl implements TransportHandler,
    TransportDebugFlags {

    /** Enables us to interact with the protocol layers. */
    private Handler handler;

    /** Maintains a list of the supported link types */
    private Hashtable linkTypes;

    /** A listing of the links hosted at this node */
    private Hashtable links;

    /** A listing of the alternate links available for communication */
    private Hashtable alternateLinks;

    /** A listing of all the link factories that the transport handler manages */
    private Hashtable linkFactories;

    /** A list of the broker nodes that this node is connected to */
    private Hashtable brokerNodes;

    /** A list of the client nodes that this node is connected to */
    private Hashtable clientNodes;

    private MonitoringService monitoringService;

    private MonitoringDataEvaluator monitoringDataEvaluator;

    private Hashtable supportedTransportProtocols;

    private TransmissionManager transmissionManager;

    private BrokerConnections brokerConnections;

    private String moduleName = "TransportHandlerImpl: ";

    public TransportHandlerImpl(Handler handler) {
        this.handler = handler;
        linkTypes = new Hashtable();
        links = new Hashtable();
        alternateLinks = new Hashtable();
        linkFactories = new Hashtable();
        brokerNodes = new Hashtable();
        clientNodes = new Hashtable();
        supportedTransportProtocols = new Hashtable();
        transmissionManager = new TransmissionManager();
        brokerConnections = new BrokerConnections(this);
        monitoringDataEvaluator = new MonitoringDataEvaluator(this);
    }

    /**
     * This is used to set the interval size at which tests on connection status
     * need to be performed.
     */
    public void setLinkStatusIntervals(long pingIntervals) {
        Enumeration e = links.elements();
        while (e.hasMoreElements()) {
            Link _link = (Link) e.nextElement();
            _link.setLinkStatusInterval(pingIntervals);
        }
    }

    /**
     * Sets the interval size for a connection status checks for links managed
     * by a specific link factory.
     */
    public void setLinkStatusIntervals(LinkFactory linkFactory,
                                       long pingIntervals) {
        String key = linkFactory.getManagedLinksType();
        if (linkFactories.containsKey(key)) {
            LinkFactory _factory = ((LinkFactory) linkFactories.get(key));
            _factory.setLinkStatusInterval(pingIntervals);
            return;
        }
        System.out.println(moduleName + "Could not locate linkfactory"
            + " of type [" + linkFactory.getManagedLinksType());
    }

    /** Provides a list of transport services that are supported by the node */
    public String[] getListOfCommunicationProtocolsSupported() {
        String[] linkTypes = new String[linkFactories.size()];
        Enumeration e = linkFactories.elements();
        int i = 0;
        while (e.hasMoreElements()) {
            LinkFactory _linkFactory = (LinkFactory) e.nextElement();
            linkTypes[i++] = _linkFactory.getManagedLinksType();
        }
        return linkTypes;
    }

    public void loadCommunicationsOfType(Properties props, String commType)
                                                                           throws TransportException {
        String _errorInfo;
        if (linkFactories.containsKey(commType)) {
            _errorInfo = moduleName + "Link factory for [" + commType
                + "] communications has already been loaded";
            System.out.println(_errorInfo);
            return;
        }
        LinkFactory _linkFactory = null;
        if (commType.equalsIgnoreCase("tcp")) {
            _linkFactory = new TCPLinkFactory();
        }

        if (commType.equalsIgnoreCase("ptcp")) {
            _linkFactory = new PTCPLinkFactory();
        }

        if (commType.equalsIgnoreCase("udp")) {
            _linkFactory = new UDPLinkFactory();
        }

        if (commType.equalsIgnoreCase("multicast")) {
            _linkFactory = new MulticastLinkFactory();
        }

        if (commType.equalsIgnoreCase("ssl")) {
            _linkFactory = new SSLLinkFactory();
        }

        if (commType.equalsIgnoreCase("http")) {
            _linkFactory = new HTTPLinkFactory();
        }

        if (commType.equalsIgnoreCase("https")) {
            _linkFactory = new HTTPSLinkFactory();
        }

        if (commType.equalsIgnoreCase("rtp")) {
            _linkFactory = new RTPLinkFactory();
        }

        if (commType.equalsIgnoreCase("niotcp")) {
            _linkFactory = new NIOTCPLinkFactory();
        }

        if (commType.equalsIgnoreCase("pooltcp")) {
            _linkFactory = new PoolTCPLinkFactory();
        }

        if (commType.equalsIgnoreCase("up2p")) {
            _linkFactory = new UP2PLinkFactory();
        }
        
        if (commType.equalsIgnoreCase("ipsec")) {
            _linkFactory = new IPSecLinkFactory();
        }
        
        if (_linkFactory != null) {
            loadLinkFactory(_linkFactory, props);
        } else {
            _errorInfo = moduleName + "Unable to locate LinkFactory for ["
                + commType + "] communications";
            if (TransportHandlerImpl_Debug) {
                System.out.println(_errorInfo);
            }
            throw new TransportException(_errorInfo);
        }
        /** end linkfactory "null"? */
    }

    /** Used to load a specific communications handler */
    private void loadLinkFactory(LinkFactory linkFactory,
                                 Properties factoryProperties)
                                                              throws TransportException {
        String _key = linkFactory.getManagedLinksType();
        if (_key == null) {
            String _errorInfo = moduleName + "LinkFactory type not set";
            throw new TransportException(_errorInfo);
        }

        if (linkFactories.containsKey(_key)) {
            System.out.println(moduleName
                + "LinkFactory services loaded previously");
            return;
        }

        linkFactory.setLinkFactoryProperties(factoryProperties);
        linkFactory.setTransportHandler(this);
        linkFactory.loadLinkFactoryServices();
        linkFactories.put(_key, linkFactory);

        supportedTransportProtocols.put(_key, factoryProperties);
    }

    /**
     * This method is invoked by a link factory when it has successfully set up
     * a connection based on the specific transport protocol.
     */
    public synchronized void registerLink(LinkFactory linkFactory, Link link) {
        String key = link.getLinkId();
        if (!links.containsKey(key)) {
            links.put(key, link);
            if (TransportHandlerImpl_Debug) {
                System.out.println(moduleName + "Registering link ["
                    + link.getLinkId() + "]");
            }
            return;
        }
        System.out.println(moduleName + "There already exists a link with id ["
            + key + "]");
    }

    /**
     * This method is to be invoked by link factories to report communication
     * failure in one of the links (reported in argument) that it manages
     */
    public synchronized void manageLinkLoss(Link link) {
        System.out.println(moduleName + "Managing Link Loss");

        /*
         * HG:MGMT First we need to determine if this link has an assigned
         * NaradaBrokering id
         */
        String _unassignedId = link.getLinkId();
        if (links.containsKey(_unassignedId)) {
            /**
             * simply remove the unassigned temporary link
             */
            links.remove(_unassignedId);

            // HG: Now notify the handler that the link has broker...
            if (handler instanceof CommunicationsService) {
                ((CommunicationsService) handler).processConnectionLost();
                return;
            }
            
        }

        /**
         * We need to do a check whether there are alternate paths to
         * communicate with the node is question. We also need to determine if
         * the connection is to a client or to a broker
         */
        String _linkId = link.getNaradaBrokeringId().toString();
        String _linkType = link.getLinkType();

        /**
         * Remove this link from the alternate links list, if it does figure in
         * that list
         */
        String _alternateKey = _linkId + _linkType;
        if (alternateLinks.containsKey(_alternateKey)) {
            /** We have lost communications over an alternate link. */
            alternateLinks.remove(_alternateKey);
        }

        /**
         * Check to see if the link used to be the default link to the node. If
         * it is remove it from the list of default link
         */
        if (links.containsKey(_linkId)) {
            Link _link = (Link) links.get(_linkId);
            if (_link.getLinkType().equals(_linkType)) {
                /** We need to remove this link from the links list */
                links.remove(_linkId);
            }
        }

        /**
         * Check to see if there exists an alternate path after our removals of
         * the link from the default/alternate links list
         */
        if (links.contains(_linkId) || alternateLinksAvailableTo(_linkId)) {
            /** Aternate communication link exists - Nothing to report */
            return;
        }

        String linkId = link.getLinkId();
        if (links.containsKey(linkId)) {
            links.remove(linkId);
        }

        /** This means that communication is no longer possible with this node */
        if (clientNodes.containsKey(_linkId)) {
            /** report loss of communications with a client */
            Object _clientIdentifier = clientNodes.get(_linkId);
            if (handler instanceof ProtocolHandler) {
                ((ProtocolHandler) handler)
                    .connectionToClientLost(_clientIdentifier, link.getLinkId());
            }
            clientNodes.remove(_linkId);
            return;
        } /* end if client connection lost */

        if (brokerNodes.containsKey(linkId)) {
            /** report loss of communications with a broker */
            Object _brokerConnectionId = brokerNodes.get(linkId);
            try {
                String report = "";
                Object connId = brokerConnections
                    .processLossOfConnection(linkId);
                if (_brokerConnectionId.equals(connId)) {
                    report = moduleName + "Proceeding to remove connection "
                        + "identifier [" + connId + "]";
                } else {
                    report = moduleName
                        + "Connection identifier mismatch during "
                        + "removal First:" + _brokerConnectionId
                        + ") Second: (" + connId + ")";
                }
                System.out.println(report);
            } catch (TransportException transEx) {
                System.out.println(transEx);
            }

            if (handler instanceof ProtocolHandler) {
                ((ProtocolHandler) handler)
                    .connectionToBrokerLost(_brokerConnectionId, linkId);
            }
            brokerNodes.remove(linkId);
            return;
        } /* end if broker connection lost */

    }

    /**
     * This method is to be invoked by link factory to request migration of
     * transport protocols for one of its managed links
     */
    public boolean manageLinkMigrationRequest(Link link, String migrateTo) {
        if (!linkFactories.containsKey(migrateTo)) {
            /** There is no factory that can satisfy this migration request */
            return false;
        }

        LinkFactory _linkFactory = (LinkFactory) linkFactories.get(migrateTo);

        if (!_linkFactory.canManageLinkMigrationOfType(link.getLinkType())) {
            return false;
        }
        try {
            _linkFactory.manageLinkMigration(link);
            return true;
        } catch (TransportException transEx) {
            System.out.println(transEx);
        }
        return false;
    }

    /**
     * byte[] signifies the data received over a communication link. When this
     * particular method is invoked it signifies a handshake. When a link has
     * been assigned a NaradaBrokering identifier this method will not be
     * invoked by the link.
     * 
     * Certain data pertaining to the initialization of the link, checking for
     * the status of the link etc. are not routed up to the transport handler.
     */
    public synchronized void dataReceived(byte[] data, String linkIdentifier) {
        if (monitoringDataEvaluator.isPerformanceData(data)) {
            monitoringDataEvaluator.checkLinkType(data, linkIdentifier, links);
        } else {
            // handler.handleData(new DataReceived(data, linkIdentifier));
            handler.handleData(data, linkIdentifier);
        }
    }

    /**
     * Upon receipt of data at a link that has an assigned NaradaBrokering
     * identifier, this is the method that is used.
     */
    public void dataReceived(byte[] data, Object linkIdentifier) {
        /** NBNI */
    }

    /**
     * This is used to setup a communication link between 2 NaradaBrokering end
     * points. The communication link type is specified in the arguments
     */
    public String setupLink(Properties linkProperties, String linkType)
                                                                       throws TransportException {
        if (!linkFactories.containsKey(linkType)) {
            throw new TransportException(moduleName
                + "Unable to locate appropriate factory");
        }

        LinkFactory _linkFactory = (LinkFactory) linkFactories.get(linkType);

        Link _createdLink = _linkFactory.createLink(linkProperties);
        return _createdLink.getLinkId();
    }

    /**
     * Used by the protocol layers to setup alternate links for communication
     * based on the type of data that is being routed. Such alternate links also
     * have a keepAlive timer associated with them, which ensure the garbage
     * collection of the link after the time has expired
     */
    public void setupAlternateLink(Properties linkProperties,
                                   Object naradaBrokeringId, String linkType,
                                   long keepAliveTime)
                                                      throws TransportException {
        String _alternateLinkKey = naradaBrokeringId.toString() + linkType;

        /** first test to see if there already exists an alternate link */
        if (alternateLinks.containsKey(_alternateLinkKey)) {
            return;
        }

        String linkId = setupLink(linkProperties, linkType);
        Link _link = (Link) links.get(linkId);
        if (_link == null) {
            throw new TransportException(moduleName
                + "Alternate link creation failed");
        }
        _link.setNaradaBrokeringId(naradaBrokeringId);
        _link.setKeepAliveTime(keepAliveTime);

        links.remove(linkId);
        alternateLinks.put(_alternateLinkKey, _link);

        /** Next we need to pass the naradaBrokering id of "this" node. */
    }

    /**
     * Send data to an unitialized node (or a node to which connection is
     * initiated for the first time).
     */
    public void sendData(byte[] data, String uninitializedNode)
                                                               throws TransportException {
        if (!links.containsKey(uninitializedNode))
            throw new TransportException(moduleName + "Unmapped Link ["
                + uninitializedNode + "] not found");

        Link _link = (Link) links.get(uninitializedNode);
        _link.sendData(data);
    }

    
    
    /** Used to send a stream of bytes to a NaradaBrokering node */
    public void sendData(byte[] data, Object naradaBrokeringId)
                                                               throws TransportException {
        String key = naradaBrokeringId.toString();
        if (clientNodes.containsKey(key)) {
            sendData(data, key);
            return;
        }

        brokerConnections.sendData(data, naradaBrokeringId);
    }

    /**
     * Send data to an unitialized node (or a node to which connection is
     * initiated for the first time). This method also includes information
     * regarding the type (audio/video/xml-attachments etc.) of the data
     * encapsulated within the data stream.
     */
    public void sendData(byte[] data, String uninitializedNode, String dataType)
                                                                                throws TransportException {
        if (!links.containsKey(uninitializedNode))
            throw new TransportException(moduleName + "Link ["
                + uninitializedNode + "] not found");

        Link _link = (Link) links.get(uninitializedNode);
        _link.sendData(data, dataType);
    }

    /**
     * Used to send a stream of bytes to a NaradaBrokering node. This method
     * also includes information regarding the type (audio/video/xml-attachments
     * etc.) of the data encapsulated within the data stream.
     */
    public void sendData(byte[] data, Object naradaBrokeringId, String dataType)
                                                                                throws TransportException {
        sendData(data, naradaBrokeringId.toString(), dataType);
    }

    /**
     * Used to send a stream of bytes to a NaradaBrokering node, using a
     * specified communication type. The alternate link should have already been
     * set up prior to the invocation of this method. This method also includes
     * information regarding the type (audio/video/xml-attachments etc.) of the
     * data encapsulated within the data stream.
     */
    public void sendData(byte[] data, Object naradaBrokeringId,
                         String linkType, String dataType)
                                                          throws TransportException {
        /** NBNI */
        String _alternateLinkKey = naradaBrokeringId.toString() + linkType;
        if (alternateLinks.containsKey(_alternateLinkKey)) {
            Link _link = (Link) alternateLinks.get(_alternateLinkKey);
            _link.sendData(data, dataType);
            return;
        }

        String _linkKey = naradaBrokeringId.toString();
        if (links.containsKey(_linkKey)) {
            Link _link = (Link) links.get(_linkKey);
            if (_link.getLinkType().equals(linkType)) {
                _link.sendData(data, dataType);
                return;
            }
        }
        throw new TransportException(moduleName + "Unable to use alternate "
            + "link [" + linkType + " for data type [" + dataType + "]");
    }

    /**
     * Enumerate the list of links available. What is returned is the linkID of
     * all the links originating from this node. Returns NULL if no links are
     * available.
     */
    public String[] enumerateLinks() {
        int _linkTotal = links.size() + alternateLinks.size();
        if (_linkTotal == 0) {
            return null;
        }

        String[] linkIds = new String[_linkTotal];
        int i = 0;
        for (Enumeration e = links.keys(); e.hasMoreElements();) {
            linkIds[i] = (String) e.nextElement();
            i++;
        }

        for (Enumeration e = alternateLinks.keys(); e.hasMoreElements();) {
            linkIds[i] = (String) e.nextElement();
            i++;
        }
        return linkIds;
    }

    /**
     * Eliminate all links associated with communicating with the specified
     * naradaBrokering id
     */
    public void closeLinks(Object naradaBrokeringId) {
        String _linkKey = naradaBrokeringId.toString();
        closeLinks(_linkKey);
    }

    /** Close all links associated with the specified Id. */
    public void closeLinks(String linkId) {
        if (links.containsKey(linkId)) {
            Link _link = (Link) links.get(linkId);
            _link.closeLink();
            links.remove(linkId);
        }

        String[] _linkTypes = getListOfCommunicationProtocolsSupported();
        if (_linkTypes == null) {
            System.out.println(moduleName + "Link Type returned is NULL");
            return;
        }
        for (int i = 0; i < _linkTypes.length; i++) {
            String _alternateKey = linkId + _linkTypes[i];
            if (alternateLinks.containsKey(_alternateKey)) {
                Link _alternateLink = (Link) alternateLinks.get(_alternateKey);
                _alternateLink.closeLink();
                alternateLinks.remove(_alternateKey);
            } /* end if */
        }
        /** end for */
    }

    public void closeAllLinks() {
        Enumeration _links = links.elements();
        while (_links.hasMoreElements()) {
            ((Link) _links.nextElement()).closeLink();
        }
        links.clear();

        Enumeration _alternateLinks = alternateLinks.elements();
        while (_alternateLinks.hasMoreElements()) {
            ((Link) _alternateLinks.nextElement()).closeLink();
        }
        links.clear();
    }

    public void disposeAllCommunicationResources() {
        /** First we need to close all links */
        closeAllLinks();
        Enumeration e = linkFactories.elements();

        while (e.hasMoreElements()) {
            ((LinkFactory) e.nextElement()).dispose();
        }
    }

    /**
     * This method is invoked from the protocol layers, after assigning a
     * NaradaBrokering id
     */
    public synchronized void mapLinkToClientNode(String linkId,
                                                 Object naradaBrokeringId,
                                                 Object clientIdentifier) {
        String _key = naradaBrokeringId.toString();

        if (clientNodes.containsKey(_key)) {
            System.out.println(moduleName
                + "There already exists a connection to" + "this client node");
            return;
        }

        if (!links.containsKey(linkId)) {
            System.out.println(moduleName + "No idea about link [" + linkId
                + "] to client");
            return;
        }

        /** This (clientId) helps us in garbage collection of matching tables */
        clientNodes.put(_key, clientIdentifier);
        if (TransportHandlerImpl_Debug) {
            System.out.println(moduleName + "Client " + naradaBrokeringId
                + " is mapped to =>" + linkId);
        }
        /**
         * Set the links NaradaBrokering ID and ensure that the link can be
         * directly referred to by its NaradaBrokering Id
         */
        Link _link = (Link) links.get(linkId);
        _link.setNaradaBrokeringId(naradaBrokeringId);
        links.remove(linkId);
        links.put(_key, _link);
    }

    /**
     * This method is invoked from the protocol layers, after assigning a
     * NaradaBrokering id
     */
    public void mapLinkToBrokerNode(String linkId, Object naradaBrokeringId,
                                    Object brokerConnectionIdentifier) {
        System.out.println(moduleName + "Mapping Link Id " + linkId);
        if (!links.containsKey(linkId)) {
            System.out.println(moduleName + "No idea about link [" + linkId
                + "] to broker [" + naradaBrokeringId + "]");
            return;
        }

        try {
            brokerConnections.mapLinkToBrokerNode(linkId, naradaBrokeringId,
                                                  brokerConnectionIdentifier);
            /** If there are no problems, we proceed to do some housekeeping */
            brokerNodes.put(linkId, brokerConnectionIdentifier);

            Link _link = (Link) links.get(linkId);
            _link.setNaradaBrokeringId(naradaBrokeringId);

        } catch (TransportException transEx) {
            System.out.println(transEx);
        }
    }

    public boolean alternateLinksAvailableTo(String linkId) {
        String[] _linkTypes = getListOfCommunicationProtocolsSupported();
        if (_linkTypes == null) {
            System.out.println(moduleName + "Link Type returned is NULL");
            return false;
        }

        for (int i = 0; i < _linkTypes.length; i++) {
            String _alternateKey = linkId + _linkTypes[i];
            if (alternateLinks.containsKey(_alternateKey)) {
                return true;
            } /* end if */
        }
        /** end for */

        return false;
    }

    public PerformanceMeasurement getMeasurementService() {
        return null;
    }

    public Hashtable getRegisteredLinks() {
        return links;
    }

    public void shutdown() {
        for (Enumeration e = linkFactories.elements(); e.hasMoreElements();) {
            LinkFactory linkFactory = (LinkFactory) e.nextElement();
            linkFactory.dispose();
        }
        linkFactories.clear();
        transmissionManager.reset();
    }

    public void startMonitoringService() throws TransportException {
        try {
            monitoringService = new MonitoringServiceImpl();
            monitoringService.setTransportHandler(this);
            monitoringService.findMeasuringLinks();
            monitoringService.getMeasurementInitiator()
                .setMeasurementInterval(1000);
            monitoringService.startMeasurementInitiator();
        } catch (MonitoringServiceException mse) {
            throw new TransportException(moduleName + mse);
        }
    }

    public MonitoringService getMonitoringService() {
        return monitoringService;
    }

    /**
     * Gets the list of registered transport protocols along with the property
     * list associated with these protocols.
     */
    public Hashtable getSupportedTransportProtocols() {
        return supportedTransportProtocols;
    }

    public static void main(String[] args) {
        Handler protocolHandler = null;
        TransportHandlerImpl transportHandler = new TransportHandlerImpl(
                                                                         protocolHandler);

        java.io.BufferedReader r = new java.io.BufferedReader(
                                                              new java.io.InputStreamReader(
                                                                                            System.in));
        String cmd;
        StringTokenizer tok;

        try {
            System.out.println("\nType h for usage options");
            while (true) {
                cmd = r.readLine();

                if (cmd == null) {
                    transportHandler.disposeAllCommunicationResources();
                    break;
                }

                if (cmd.startsWith("h")) {
                    System.out.println("ld  <port> <type> \t \t loads factory");
                    System.out.println("cl <link-id> \t \t \t Closes a link");
                    continue;
                }

                if (cmd.startsWith("ld")) {
                    tok = new StringTokenizer(cmd, " ");
                    if (tok.countTokens() != 3) {
                        System.out
                            .println("ld <port> <type> \t \t loads factory");
                        continue;
                    }
                    Properties _props = new Properties();
                    tok.nextToken();
                    _props.put("TCPServerPort", tok.nextToken());
                    transportHandler.loadCommunicationsOfType(_props, tok
                        .nextToken());
                }
                /** if (creation of link */

                if (cmd.equalsIgnoreCase("q")) {
                    transportHandler.disposeAllCommunicationResources();
                    break;
                }
            }/* end while(true) */
        } catch (java.io.IOException ioe) {
            ioe.printStackTrace();
        } catch (TransportException transEx) {
            System.out.println(transEx);
        }/* end try-catch */

    } /* end (main) */

    /**
     * @see cgl.narada.transport.TransportHandler#getTransmissionManager()
     */
    public TransmissionManager getTransmissionManager() {
        return transmissionManager;
    }

}
